package com.aoip.telemetry.telemetry.analysis.decoder;

import com.aoip.telemetry.telemetry.core.redis.RedisCache;
import com.aoip.telemetry.telemetry.entity.enums.CacheKeyEnum;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 从rabbitmq中消费报文数据信息
 *
 * @author chenbai
 * @version 1.0
 * @date 2022/10/13 14:10
 */
@Component
@Slf4j
public class ConsumeRabbitMq {

    @Resource
    private MsgDecoder decode;

    @Resource
    private RedisCache redisCache;

    @Value("${aoip.telemetry.analysis.check-repeat}")
    private boolean checkRepeat;

    private Lock lock = new ReentrantLock();

    @RabbitHandler
    @RabbitListener(queues = {"queue_hydrology"})
    public void consume(String msg, Channel channel, Message message) {
        msg = msg.trim();
        if (!StringUtils.hasText(msg)) {
            return;
        }
        //防止重复解析
        if (checkRepeat) {
            String key = CacheKeyEnum.ORIGINAL_MSG + ":" + msg;
            lock.lock();
            try {
                if (redisCache.hasKey(key)) {
                    return;
                }
                redisCache.setCacheObject(key, "ORIGINAL_MSG", 1, TimeUnit.HOURS);
            } finally {
                lock.unlock();
            }
        }
        /**
         * 确认一条消息：<br>
         * channel.basicAck(deliveryTag, false); <br>
         * deliveryTag:该消息的index <br>
         * multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息 <br>
         */
        try {
            log.error("begin::msg:{}", msg);
            decode.decode(msg);
            //channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.error("end::msg:{}", msg);
        } catch (Exception e) {
            //消费者处理出了问题，需要告诉队列信息消费失败
            /**
             * 拒绝确认消息:<br>
             * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ; <br>
             * deliveryTag:该消息的index<br>
             * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。<br>
             * requeue：被拒绝的是否重新入队列 <br>
             */
            log.error(CacheKeyEnum.MQ_CONSUME_FAIL.getCode() + ":" + msg, e.getMessage());
            //try {
              //  channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
                redisCache.setCacheObject(CacheKeyEnum.MQ_CONSUME_FAIL.getCode() + ":" + msg, e.getMessage());
            //} catch (IOException ex) {
            //  log.error("消息消费失败：", e.fillInStackTrace());
            //    ex.printStackTrace();
            //}
            /**
             * 拒绝一条消息：<br>
             * channel.basicReject(long deliveryTag, boolean requeue);<br>
             * deliveryTag:该消息的index<br>
             * requeue：被拒绝的是否重新入队列
             */
            //channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }

    }
}
